package day06

import java.lang

import org.apache.spark.{SparkConf, SparkContext}
import org.apache.spark.rdd.RDD
import org.apache.spark.util.{DoubleAccumulator, LongAccumulator}

object AccumulatorDemo2 {
  def main(args: Array[String]): Unit = {
    val conf: SparkConf = new SparkConf().setAppName("accu").setMaster("local[*]")
    val sc = new SparkContext(conf)

    val number1: RDD[Int] = sc.parallelize(List(1,2,3,4,5,6),2)
    val number2: RDD[Double] = sc.parallelize(List(1.1, 2.2, 3.5, 4.6, 5.7), 2)

    // 创建并且注册累加器。这里使用spark新版给定好的累加器
    def longAcc(name: String): LongAccumulator = {
      val acc = new LongAccumulator
      sc.register(acc, name)
      return acc
    }
    val acc = longAcc("longacc")
    number1.foreach(acc.add(_))

    // 浮点数的累加器
    // 创建并且注册累加器。这里使用spark新版给定好的累加器
    def DoubleAcc(name: String): DoubleAccumulator = {
      val acc = new DoubleAccumulator
      sc.register(acc, name)
      return acc
    }
    val acc2 = DoubleAcc("doubleacc")
    number2.foreach(acc2.add(_))

    println(acc.value)
    println(acc2.value)
    sc.stop()

  }
}
